package com.shujia.spark.streaming

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo06StructuredStreaming {
  def main(args: Array[String]): Unit = {
    // 基于DF的流式查询，本质上还是微批处理，延时很低，批次的大小是100ms，默认是有状态的

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 至少需要两个线程，因为接收数据需要一致占用一个线程
      .config("spark.sql.shuffle.partitions", "1")
      .getOrCreate()

    val lineDF: DataFrame = spark
      .readStream
      .format("socket")
      .option("host", "master")
      .option("port", 8888)
      .load()

    lineDF.createOrReplaceTempView("words")

    val wordCnfDF: DataFrame = spark.sql(
      """
        |select  t1.word
        |       ,count(*) as cnt
        |from (
        |select explode(split(value,",")) as word
        |from words
        |) t1 group by t1.word
        |""".stripMargin)

    //    wordCnfDF.show() // 离线的DF操作可以通过show触发任务

    wordCnfDF
      .writeStream
      /**
      * append：只支持不含聚合操作的输出，会将新来的数据追加输出
      * complete：支持包含聚合操作的输出，会将所有的聚合结果进行输出
      * update：只输出有更新的结果，如果不含聚合操作，和append效果一样
      */
      .outputMode("complete")
      //      .format("console") // 通过控制台打印
      .foreachBatch((df:DataFrame,l:Long)=>{
        df
          .write
          .mode(SaveMode.Overwrite)
          .format("jdbc")
          .option("url", "jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32")
          .option("dbtable", "bigdata32.word_cnt_zzk")
          .option("user", "bigdata32")
          .option("password", "123456")
          .save()
      })
      // 在流式DF上需要通过writeStream后通过start及awaitTermination启动任务
      .start()
      .awaitTermination()
  }

}
